package yuan.topic;


import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @version: 1.0
 * @Description: TODO
 * @author: zZmH
 * @date: 2020/09/09 15:06
 */


public class TopicCommand {
    static String brokerList = "192.168.190.128:9092";


    public static Properties init(){
        final Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);


        return properties;
    }


    public static void describeTopic(){
        Properties prop = init();

        AdminClient client = AdminClient.create(prop);

        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, "topic-demo2");
        DescribeConfigsResult result = client.describeConfigs(Collections.singleton(configResource));
        try {
            Map<ConfigResource, Config> configMap = result.all().get();

            System.out.println(configMap);
            client.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }




    }

    public static void createTopic(String topic){
        final Properties props = init();

        AdminClient client = AdminClient.create(props);
        final NewTopic newTopic = new NewTopic(topic, 1, (short) 1);
        CreateTopicsResult result = client.createTopics(Collections.singleton(newTopic));

        try {
            result.all().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        client.close();


    }

    public static void updateTopic(String topic) {
        Properties prop = init();
        AdminClient client = AdminClient.create(prop);
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        ConfigEntry entry = new ConfigEntry("cleanup.policy", "compact");
        Config config = new Config(Collections.singleton(entry));
        HashMap<ConfigResource, Config> configResourceConfigHashMap = new HashMap<>();
        configResourceConfigHashMap.put(configResource, config);
        AlterConfigsResult result = client.alterConfigs(configResourceConfigHashMap);

        try {
            result.all().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }



    public static void main(String[] args) {
//        create("topic-demo2");


//        describeTopic();

    }




}
